package storm;


import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.PrintStream;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import com.aliasi.tokenizer.PorterStemmerTokenizerFactory;

import utilities.*;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

public class StemmerBolt extends BaseBasicBolt {

	private static final long serialVersionUID = 4493364795064805398L;
	private List<String> list = new LinkedList<String>();
	
	@Override
	public void execute(Tuple input, BasicOutputCollector collector) {
		String tweet = input.getString(0);
		Twokenizer tokenizer = new Twokenizer();
		List<String> tokens = tokenizer.twokenize(tweet);
		String tmp;
		String stem;
		String tweetStem = "";
		Iterator<String> iter = tokens.iterator();
		while(iter.hasNext()){
			tmp = ((String)iter.next());
			if(tmp.startsWith("#")||tmp.startsWith("@")||tmp.contains("http")){
				tweetStem = tweetStem + " " + tmp;
			}
			else {
				stem = PorterStemmerTokenizerFactory.stem(tmp);
				tweetStem = tweetStem + " " + stem;
			}
		}
		list.add(tweetStem);
		cleanup();
	}
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("tweetStem"));
	}

	public void cleanup() {
		createFile();
				System.out.println("CREATED DUMP FILE");
		
	}
	private void createFile(){
		PrintStream ps=null;
		try {
			//ps = new PrintStream(new FileOutputStream("android.txt"));
			//ps = new PrintStream(new FileOutputStream("jobs.txt"));
			//ps = new PrintStream(new FileOutputStream("love.txt"));
			//ps = new PrintStream(new FileOutputStream("ipad.txt"));
			ps = new PrintStream(new FileOutputStream("output.txt"));
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		for(String s : list){
			ps.println(s);
		}
		ps.close();
	}
}
